package com.lagou;

import com.lagou.boot.ConsumerBoot;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;


@SpringBootApplication
public class ConsumerBootstrap {

    public static void main(String[] args) throws Exception {
        SpringApplication.run(ConsumerBootstrap.class,args);
        connetZKProvider();
    }

    public static void connetZKProvider() throws Exception {
        RetryPolicy exponentialBackoffRetry = new ExponentialBackoffRetry(1000, 3);
        // 使用fluent编程风格
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString("127.0.0.1:2181")
                .sessionTimeoutMs(50000)
                .connectionTimeoutMs(30000)
                .retryPolicy(exponentialBackoffRetry)
                .namespace("rpc")  // 独立的命名空间 /base
                .build();

        client.start();
        System.out.println("客户端ZK连接成功");

        //遍历ZK 服务端节点，并与对应节点连接
        String path = "/rpc-provider";
        List<String> serviceNode = client.getChildren().forPath(path);
        //创建线程池对象
        ExecutorService executor = Executors.newCachedThreadPool();

        for (String spath : serviceNode) {
            byte[] bytes = client.getData().forPath("/rpc-provider/"+spath);
            String[] ipPort = new String(bytes).split(":");
            System.out.println("获取到的服务端节点IP和端口：" + ipPort[0]+":"+ipPort[1]);

            ConsumerBoot consumerBoot = new ConsumerBoot(ipPort[0] ,Integer.parseInt(ipPort[1]));
            executor.submit(consumerBoot);

        }
        setListener(client);
    }

    public static void setListener(CuratorFramework client) throws Exception {
        PathChildrenCache cache =
                new PathChildrenCache(client, "/rpc-provider", true);
        PathChildrenCacheListener l =
                new PathChildrenCacheListener() {
                    @Override
                    public void childEvent(CuratorFramework client,
                                           PathChildrenCacheEvent event) {
                        try {
                            ChildData data = event.getData();
                            switch (event.getType()) {
                                case CHILD_ADDED:
                                    System.out.println("子节点增加, path={}, data={}"+ data.getPath());
                                    connetZKProvider();
                                    break;
                                case CHILD_UPDATED:
                                    System.out.println("子节点删除, path={}, data={}"+ data.getPath());
                                    break;
                                case CHILD_REMOVED:
                                    System.out.println("子节点改动, path={}, data={}"+ data.getPath());
                                    break;
                                default:
                                    break;
                            }

                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                };
        cache.getListenable().addListener(l);
        cache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
    }


}
